from pyflink.common import Duration, WatermarkStrategy,Time
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.file_system import FileSource, StreamFormat
from pyflink.datastream.window import TumblingEventTimeWindows, TumblingProcessingTimeWindows

env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

#2、监控目录，构建无界流
#for_record_stream_fromat：指定读取数据的格式和路径
#monitor_continuously:每隔一段时间扫描目录下新的文件，构建无界流
file_source = (FileSource.for_record_stream_format(StreamFormat.text_line_format('UTF-8'),
                                              '../../data/words').monitor_continuously(Duration.of_millis(5)).build())

lines_ds = env.from_source(file_source, WatermarkStrategy.no_watermarks(), 'file_source')

key_by_ds = lines_ds.flat_map(lambda line: line.split(',')).map(lambda word: (word, 1)).key_by(lambda kv: kv[0])
# 划分窗口，每隔10秒一个窗口

window_ds = key_by_ds.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
# 在窗口内统计单的数量

count_ds = window_ds.reduce(lambda value1, value2: (value1[0], value1[1] + value2[1]))

count_ds.print()

env.execute()

